package cn.ac.iie.di.ban.nmpp.handler;

import cn.ac.iie.di.ban.nmpp.commons.RuntimeEnv;
import cn.ac.iie.di.datadock.rdata.exchange.client.core.REFieldType;
import com.datastax.driver.core.*;
import com.datastax.driver.core.policies.DCAwareRoundRobinPolicy;
import com.datastax.driver.core.policies.DefaultRetryPolicy;
import org.apache.commons.codec.DecoderException;
import org.apache.commons.codec.binary.Hex;
import org.codehaus.jackson.map.ObjectMapper;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.*;

import org.apache.log4j.Logger;
import static cn.ac.iie.di.ban.nmpp.commons.RuntimeEnv.UPDATE_FORWARDMQ;
import static com.datastax.driver.core.DataType.Name.SET;
import com.datastax.driver.core.exceptions.InvalidQueryException;
import com.datastax.driver.core.exceptions.NoHostAvailableException;
import com.datastax.driver.core.exceptions.OperationTimedOutException;
import com.datastax.driver.core.exceptions.QueryExecutionException;
import static java.lang.Thread.sleep;

/**
 * Created by jinsheng on 16/9/18.
 *
 */
public class SE2DWWorkercommonUpdateE extends SE2DWWorker {

    private String tableName;
    private String utableName;
    private String timestampKey = "";
    private String keySpace;
    long startTime = -1;
//rocketmq Type
    private Map<String, REFieldType> rtn;
    private List<String> updateFields;
    static Map<String, String> ColumnNameType = new HashMap<String, String>();
    private List<ColumnMetadata> columns;
    private List<ColumnMetadata> columns2;
    private Map<String, Type> columnMap = new HashMap<>();
    private Map<String, Type> columnMap2 = new HashMap<>();

    private final ObjectMapper objectMapper = new ObjectMapper();
    private static org.apache.log4j.Logger logger = Logger.getLogger(SE2DWWorkercommonUpdateE.class.getName());

    private Cluster cluster;
    private Session session;
//    updateInsert
    private Cluster clusterUI;
    private Session sessionUI;

    private String updatePrimaryKey;
    private PreparedStatement insert_statement = null;
    private PreparedStatement update_statment = null;

    public SE2DWWorkercommonUpdateE(TableSe2DWHandler pTableSe2DWHandler, String pDWType) {
        super(pTableSe2DWHandler, pDWType);
    }

    public List<ColumnMetadata> getColumns() {
        return columns;
    }

    @Override
    public void init() throws Exception {
        tableName = tableSe2DBRule.getTableName();
        if (UPDATE_FORWARDMQ.get(tableName) != null) {
            updateFields = UPDATE_FORWARDMQ.get(tableName).getColumnSet();
            updatePrimaryKey = UPDATE_FORWARDMQ.get(tableName).getQueryid();
            logger.info(" " + updatePrimaryKey + " " + updateFields);
        }
        initPStatement();
        uiPStatement();
    }

    private void initConnection() {

        if (cluster != null) {
            return;
        }
        int readTimeOut = 12000;
        Boolean retry = true;
        for (int tryTimes = 0;; tryTimes++) {
            try {
                logger.info(tryTimes + "st times try get available connection to " + tableSe2DBRule.getDbType() + " dw");
                String[] dbmeta = tableSe2DBRule.getConnStr().split("[/]");
                if (dbmeta.length != 3) {
                    logger.error("unknow dbtype " + tableSe2DBRule.getDbType() + " dw unsuccessfully for " + tableSe2DBRule.getConnStr());
                }
                String userName = tableSe2DBRule.getUserName();
                String password = tableSe2DBRule.getPassword();

                QueryOptions options = new QueryOptions();
                options.setConsistencyLevel(ConsistencyLevel.LOCAL_ONE);
                Cluster.Builder builder = Cluster.builder();
                String dataCenter = dbmeta[2];
                keySpace = dbmeta[1];
                tableName = tableSe2DBRule.getTableName();
//                String coordinatorAddress = dbmeta[0];
                String [] Address = dbmeta[0].split("[;]");
                Random ran = new Random();
                int num = ran.nextInt(Address.length);
                String coordinatorAddress = Address[num];
                //创建连接
                //需要用户认证
                if (userName.length() > 0) {
                    builder
                            .withSocketOptions(new SocketOptions().setConnectTimeoutMillis(120000).setReadTimeoutMillis(readTimeOut))
                            .withLoadBalancingPolicy(DCAwareRoundRobinPolicy.builder().withLocalDc(dataCenter).build())
                            //.withRetryPolicy(DefaultRetryPolicy.INSTANCE)
                            .withCredentials(userName, password).addContactPoint(coordinatorAddress).withQueryOptions(options);
                    // 不需要用户认证{
                } else {
                    builder
                            .withSocketOptions(new SocketOptions().setConnectTimeoutMillis(120000).setReadTimeoutMillis(readTimeOut))
                            .withLoadBalancingPolicy(DCAwareRoundRobinPolicy.builder().withLocalDc(dataCenter).build())
                            //.withRetryPolicy(DefaultRetryPolicy.INSTANCE)
                            .addContactPoint(coordinatorAddress).withQueryOptions(options);
                }
                // .withCredentials(username, passwd).addContactPoint(ip).withQueryOptions(options);
                if (retry == null || retry) {
                    this.cluster = builder.withRetryPolicy(DefaultRetryPolicy.INSTANCE).build();
                } else {
                    this.cluster = builder.build();
                }
                break;
            } catch (Exception e) {
                logger.error("cluser create error "+e.getMessage(),e);
                try {
                    cluster.close();
                    sleep(500);
                } catch (Exception ex) {
                    logger.error(ex.getMessage(),ex);
                }finally{
                    cluster = null;
                }
            }
        }

    }

    private void uiPStatement() {
        if (clusterUI != null || sessionUI != null || insert_statement != null) {
            return;
        }

        for (int tryTimes = 0;; tryTimes++) {
            try {
                String UI = UPDATE_FORWARDMQ.get(tableName).getForwardMq();
                String dc = "";
                String ks = "";
                String tb = "";
                String ip = "";
                String username = "";
                String passwd = "";
                String[] updateTableConf = UI.split(":");
                if (updateTableConf.length == 5) {
                    String [] Address = updateTableConf[0].split("[;]");
                    Random ran = new Random();
                    int num = ran.nextInt(Address.length);
                    ip = Address[num];
//                    ip = updateTableConf[0];
                    ks = updateTableConf[1];
                    dc = updateTableConf[2];
                    utableName = updateTableConf[3];
                    if (updateTableConf[4].split(";").length != 2) {
                        logger.info("update sname is error" + updateTableConf[4]);
                        System.exit(0);
                    }
                    username = updateTableConf[4].split(";")[0];
                    passwd = updateTableConf[4].split(";")[1];
                    tb = updateTableConf[3];
//            10.136.131.130:data_update:data-update:tp_msg_wechat_public_interactive_ulog:cassandra;cassandra
                } else {
                    logger.info("update table is null");
                    System.exit(0);

                }
                int readTimeOut = 12000;
                Boolean retry = true;
                //�创建链接
                QueryOptions options = new QueryOptions();
                options.setConsistencyLevel(ConsistencyLevel.LOCAL_ONE);
                Cluster.Builder builder = Cluster.builder();

                builder
                        .withSocketOptions(new SocketOptions().setConnectTimeoutMillis(120000).setReadTimeoutMillis(readTimeOut))
                        .withLoadBalancingPolicy(DCAwareRoundRobinPolicy.builder().withLocalDc(dc).build())
                        //.withRetryPolicy(DefaultRetryPolicy.INSTANCE)
                        .withCredentials(username, passwd).addContactPoint(ip).withQueryOptions(options);

                //设置重试策略�
                if (retry == null || retry) {
                    this.clusterUI = builder.withRetryPolicy(DefaultRetryPolicy.INSTANCE).build();
                } else {
                    this.clusterUI = builder.build();
                }
//绑定keyspace
                this.sessionUI = this.clusterUI.connect(ks);
                //获取该表元数据Ԫ���
                this.columns2 = this.clusterUI.getMetadata().getKeyspace(ks).getTable(utableName).getColumns();
                for (ColumnMetadata name : columns2) {
                    columnMap2.put(name.getName(), resolveType(name.getType()));
                }
                logger.info(columnMap2.toString());
                if (insert_statement == null) {
                    // 预编译insert语句
                    insert_statement = sessionUI.prepare("INSERT INTO " + tb + " JSON :json");
                    logger.info("INSERT INTO " + tb + " JSON :json");
                }
                break;
            } catch (Exception E) {
                logger.error(E.getMessage(), E);
                uclose();
            }
        }
    }

    public void initPStatement() {
        long stinit = System.currentTimeMillis();
        if (cluster == null) {
            for (int tryTimes = 0;; tryTimes++) {
                logger.info(tryTimes + "st times try get valid pstmt to table " + tableSe2DBRule.getTableName() + " on " + tableSe2DBRule.getDbType() + " dw");
                initConnection();
                if (cluster == null) {
                    logger.error("fatal error when getting connection to " + tableSe2DBRule.getDbType() + " dw");
                } else {
                    try {
                        logger.info("create pstmt...");
                        //获取该表的元数据����
                        this.session = this.cluster.connect(keySpace);
                        //处理每一列的类型并保存，用于更新时确定调用哪个set方法
                        logger.info(keySpace + " " + tableName);
                        this.columns = this.cluster.getMetadata().getKeyspace(keySpace).getTable(tableSe2DBRule.getTableName()).getColumns();
                        for (ColumnMetadata name : columns) {
                            columnMap.put(name.getName(), resolveType(name.getType()));
                        }
                        logger.info(columnMap.toString());
                        long etinit = System.currentTimeMillis();
                        logger.info("create pstmt successfully using " + (etinit - stinit) + " ms");
                        break;
                    } catch (Exception ex) {
                        logger.error("create pstmt to " + tableSe2DBRule.getTableName() + " unsuccessfully for " + ex.getMessage(), ex);
                        close();
                    }
                }
            }
        }
    }

    @Override
    public void loadData0() {
        try {
            if (docRecordSet.size() < 1) {
                logger.warn("no record load to table " + tableSe2DBRule.getTableName() + " in " + dwType);
                return;
            }
            int tryTimes = 0;
            for (tryTimes = 0; tryTimes < 10; tryTimes++) {
                logger.info(tryTimes + "st try to write " + docRecordSet.size() + " records to " + tableSe2DBRule.getTableName() + "...");
                initPStatement();
                logger.info(tryTimes + "st set parameters...");
                String rocketMqColumnName = "";
                ArrayList<Map<String, Object>> DataR = new ArrayList<Map<String, Object>>();
                if (rtn == null) {
                    rtn = RuntimeEnv.ROCKETMQ_DOC_DESC.get(tableName);
                }
                startTime = System.currentTimeMillis();
                for (HashMap docRecord : docRecordSet) {
                    Map<String, Object> newSingleData = new HashMap<String, Object>();
                    for (String columnName : updateFields) {
                        rocketMqColumnName = GetColumnName(columnName);
                        Object value = docRecord.get(rocketMqColumnName);
                        if("m_update_time".equals(rocketMqColumnName)||"u_update_time".equals(rocketMqColumnName)||"ut".equals(columnName)||"uut".equals(columnName)){
                            newSingleData.put(columnName, System.currentTimeMillis() / 1000);
                        }else if (value != null) {
                            newSingleData.put(columnName, value);
                        }
                    }
                    DataR.add(newSingleData);
                }
                logger.debug("data begin loader newmpp and size is " + DataR.size());
                try {
                    while (!updateData(DataR)) {
                        logger.error("insert error  and sleep" + 500 + "ms");
                        close();
                        initPStatement();
                    };
                } catch (Exception E) {
                    logger.error(E.getMessage(), E);
                }
                docRecordSet.clear();
                break;
            }
        } catch (Exception E) {
            logger.error(E.getMessage(), E);

        }
    }

    private boolean updateData(List<Map<String, Object>> lists) throws InterruptedException {
        List<Map<String, Object>> succ = new ArrayList<Map<String, Object>>();
        succ.addAll(lists);
        if (lists.size() < 1) {
            return true;
        }
        List<ResultSetFuture> future = new ArrayList<>();
        for (Map<String, Object> map : lists) {
            if(map.size()<1){
                continue;
            }
            try {
            // 预编译update语句
            Set<String> feilds = map.keySet();
//            if (update_statment == null) {
                String sql = getUpdatesql(updatePrimaryKey, feilds);
                logger.debug("update sql is  " + sql);
                logger.debug(map.toString());
                update_statment = session.prepare(sql);
//            }
            //绑定更新数据，并创建BoundStatement对象
            BoundStatement bs = update_statment.bind();
            setUpdateValue(map, bs, feilds);
            // 绑定更新记录的主�?
            columnMap.get(updatePrimaryKey).bind("key", map.get(updatePrimaryKey), bs);
            Integer seqn = 0;
            columnMap.get(seqn).bind("seqn", seqn, bs);//news
            // 设置记录时间�?
            if (this.timestampKey.length() > 0 && map.containsKey(this.timestampKey)) {
                bs.setDefaultTimestamp((Long) map.get(this.timestampKey) * 1000000); // second to microsecond
            }
                future.add(session.executeAsync(bs));
            } catch (Exception E) {
                logger.error(E.getMessage(), E);
                return false;
            }
        }
        // 处理执行结果
        int update = 0;
        int trytimes = 0;
        for (int i = 0; i < future.size(); i++) {
            try {
                ResultSet rs = future.get(i).get();
                String result_string = rs.toString();
                if (!result_string.contains("ERROR") && rs.wasApplied()) {
                    update++;
                } else {
                    logger.info(" update fail "+lists.get(i));
                }
            }catch (OperationTimedOutException e) {
                    logger.error("Exception " + e.getMessage(), e);
                    i--;
                    trytimes++;
                    sleep(50);
                    if (trytimes >= 5) {
                        trytimes = 0;
                        return false;
                    }
            } catch (Exception e) {
                logger.error(e.getMessage(), e);
                return false;
                // 异常捕捉�?30~234�?
            }
        }
        future.clear();
        Long et = System.currentTimeMillis();
        logger.info("0st load " + lists.size() + "|expect: " + update + " records update to table " + tableSe2DBRule.getTableName() + " and mqName is " + tableSe2DBRule.getMqName() + " successfully using " + (et - startTime) + " ms ");
        startTime = System.currentTimeMillis();
        while (!push_single_json_data(succ)) {
            logger.error("insert error  and sleep" + 500 + "ms");
            uclose();
            uiPStatement();
        };
        succ.clear();
        return true;
    }

    private String GetColumnName(String getMongodbName) throws IOException, InterruptedException {
        HashMap<String, String> h = RuntimeEnv.TABLE_CR_COLUMNS.get(tableSe2DBRule.getRuleName());

        if (h.get(getMongodbName) != null) {
            return h.get(getMongodbName);
        } else {
            return null;
        }
    }

    private boolean push_single_json_data(List<Map<String, Object>> jsonList) {
        try {
            int trytimes = 0;
            long pp = 0l;
            logger.debug("jsonList  size is " + jsonList.size());
            BoundStatement bs;
            List<ResultSet> future = new ArrayList<>();
//            List<Map<String, Object>> errjson = new ArrayList<Map<String, Object>>();
            for (int i = 0; i < jsonList.size(); i++) {
                Map<String, Object> str = jsonList.get(i);
                // 绑定数据并创建BoundStatement对象
                String json = objectMapper.writeValueAsString(str);
                bs = insert_statement.bind(json);
                // 设置记录时间�?
                if (this.timestampKey.length() > 0 && ((Map) str).containsKey(this.timestampKey)) {
                    bs.setDefaultTimestamp((Long) ((Map) str).get(this.timestampKey) * 1000000); // second to microsecond
                }//            batchStatement.add(bs);
                // 同步插入
                try {
                    ResultSet result = sessionUI.execute(bs);
                    String result_string = result.toString();
                    if (!result_string.contains("ERROR") && result.wasApplied()) {
                        pp++;
                    } else {
                        logger.info(" insert fail "+jsonList.get(i));
                    }
                } catch (OperationTimedOutException e) {
                    logger.error("Exception " + e.getMessage(), e);
                    i--;
                    trytimes++;
                    if (trytimes >= 5) {
                        return false;
                    }
                } catch (InvalidQueryException | QueryExecutionException | NoHostAvailableException e) {
                    logger.error("Exception " + e.getMessage(), e);
                    // 此处有可能是客户端sql语句不正�?
                    return false;
                } catch (Exception E) {
                    logger.error("Exception " + E.getMessage(), E);
                    return false;
                }

            }
            long et = System.currentTimeMillis();
            logger.info("0st load " + jsonList.size() + "|xxpect: " + pp + " records to table " + utableName + " successfully using " + (et - startTime) + " ms errorSIze: ");
            pp = 0l;
            future.clear();
        } catch (Exception E) {
            logger.error(E.getMessage(), E);
            return false;
        }
        return true;
    }

    private Type resolveType(DataType rawType) {
        switch (rawType.getName()) {
            case ASCII:
            case VARCHAR:
            case TEXT:
                return Type.STRING;
            case TINYINT:
                return Type.TINYINT;
            case SMALLINT:
                return Type.SMALLINT;
            case INT:
                return Type.INT;
            case BIGINT:
            case VARINT:
                return Type.BIGINT;
            case FLOAT:
                return Type.FLOAT;
            case DOUBLE:
                return Type.DOUBLE;
            case BLOB:
                return Type.BLOB;
            case BOOLEAN:
                return Type.BOOLEAN;
            case LIST:
                return Type.LIST;
            case MAP:
                return Type.MAP;
            case SET:
                return Type.SET;
            default:
                throw new IllegalArgumentException("Unsupported type: " + rawType);

        }
    }

    private enum Type {
        STRING {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else {
                    bs.setString(col, val.toString());
                }
            }
        }, TINYINT {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else if (val instanceof String) {
                    bs.setByte(col, Long.valueOf(((String) val).toLowerCase()).byteValue());
                } else if (val instanceof Number) {
                    bs.setByte(col, ((Number) val).byteValue());
                } else {
                    throw new IllegalArgumentException("wrong type of " + col);
                }
            }
        }, SMALLINT {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else if (val instanceof String) {
                    bs.setShort(col, Long.valueOf(((String) val).toLowerCase()).shortValue());
                } else if (val instanceof Number) {
                    bs.setShort(col, ((Number) val).shortValue());
                } else {
                    throw new IllegalArgumentException("wrong type of " + col);
                }
            }
        }, INT {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else if (val instanceof String) {
                    bs.setInt(col, Long.valueOf(((String) val).toLowerCase()).intValue());
                } else if (val instanceof Number) {
                    bs.setInt(col, ((Number) val).intValue());
                } else {
                    throw new IllegalArgumentException("wrong type of " + col);
                }
            }
        }, BIGINT {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else if (val instanceof String) {
                    bs.setLong(col, Long.valueOf(((String) val).toLowerCase()));
                } else if (val instanceof Number) {
                    bs.setLong(col, ((Number) val).longValue());
                } else {
                    throw new IllegalArgumentException("wrong type of " + col);
                }
            }
        }, FLOAT {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else if (val instanceof String) {
                    bs.setFloat(col, Long.valueOf(((String) val).toLowerCase()).floatValue());
                } else if (val instanceof Number) {
                    bs.setFloat(col, ((Number) val).floatValue());
                } else {
                    throw new IllegalArgumentException("wrong type of " + col);
                }
            }
        }, DOUBLE {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else if (val instanceof String) {
                    bs.setDouble(col, Long.valueOf(((String) val).toLowerCase()).doubleValue());
                } else if (val instanceof Number) {
                    bs.setDouble(col, ((Number) val).doubleValue());
                } else {
                    throw new IllegalArgumentException("wrong type of " + col);
                }
            }
        }, BLOB {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else if (val instanceof String) {
                    try {
                        byte[] bytes = Hex.decodeHex(((String) val).toCharArray());
                        bs.setBytes(col, ByteBuffer.wrap(bytes));
                    } catch (DecoderException e) {
                        throw new IllegalArgumentException("wrong value of hex string, column: " + col);
                    }
                } else if (val instanceof byte[]) {
                    bs.setBytes(col, ByteBuffer.wrap((byte[]) val));
                } else if (val instanceof ByteBuffer) {
                    bs.setBytes(col, (ByteBuffer) val);
                } else {
                    throw new IllegalArgumentException("wrong type of " + col);
                }
            }
        }, BOOLEAN {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else if (val instanceof String) {
                    bs.setBool(col, Boolean.valueOf((String) val));
                } else if (val instanceof Boolean) {
                    bs.setBool(col, (Boolean) val);
                } else {
                    throw new IllegalArgumentException("wrong type of " + col);
                }
            }
        }, LIST {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else if (val instanceof List) {
                    bs.setList(col, (List<?>) val);
                } else if (val instanceof Collection) {
                    List<?> l = new ArrayList<>((Collection<?>) val);
                    bs.setList(col, l);
                } else {
                    List<Object> l = new ArrayList<>();
                    l.add(val);
                    bs.setList(col, l);
                }
            }
        }, MAP {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else if (val instanceof Map) {
                    bs.setMap(col, (Map<?, ?>) val);
                } else {
                    throw new IllegalArgumentException("wrong type of " + col);
                }
            }
        }, SET {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else if (val instanceof Set) {
                    bs.setSet(col, (Set<?>) val);
                } else if (val instanceof Collection) {
                    Set<?> s = new HashSet<>((Collection<?>) val);
                    bs.setSet(col, s);
                } else {
                    Set<Object> s = new HashSet<>();
                    s.add(val);
                    bs.setSet(col, s);
                }
            }
        };

        public void bind(String col, Object val, BoundStatement bs) {
            throw new IllegalArgumentException("wrong type of " + col);
        }
    }

    /**
     * 提交更新数据到Cassandra
     *
     * @param lists 数据列表
     */
    public void close() {
        try {
            if (session != null) {
                this.session.close();
            }
            if (cluster != null) {
                this.cluster.close();
            }
            sleep(1000);
        } catch (Exception E) {
            logger.error(E.getMessage(), E);
        } finally {
            session = null;
            cluster = null;
//            insert_statement = null;
            update_statment = null;
        }
    }

    public void uclose() {
        try {
            if (sessionUI != null) {
                this.sessionUI.close();
            }
            if (clusterUI != null) {
                this.clusterUI.close();
            }
            sleep(1000);
        } catch (Exception E) {
            logger.error(E.getMessage(), E);
        } finally {
            sessionUI = null;
            clusterUI = null;
            insert_statement = null;
        }
    }

    /**
     * 映射Cassandra类型到{@linkplain Type}枚举
     *
     * @param rawType Cassandra类型对象
     * @return 返回Cassandra类型对应的Type枚举
     */
    public Session getSession() {
        return this.session;
    }
    //去重程序  调用去重服务

    private Object resolveType(Type tt, Object o) {
        switch (tt) {
            case STRING:
                return o.toString();
            case TINYINT:
                return Type.TINYINT;
            case SMALLINT:
                return Type.SMALLINT;
            case INT:
                return Integer.valueOf(o.toString());
            case BIGINT:
                return Long.valueOf(o.toString());
            case FLOAT:
                return Type.FLOAT;
            case DOUBLE:
                return Type.DOUBLE;
            case BLOB:
                return Type.BLOB;
            case BOOLEAN:
                return Type.BOOLEAN;
            case LIST:
                return o;
            case MAP:
                return Type.MAP;
            case SET:
                return Type.SET;
            default:
                throw new IllegalArgumentException("Unsupported type: " + tt);
        }
    }

    /**
     * 设置更新条件
     *
     * @return
     */
    private String getUpdateStr(Set<String> upFields) {
        StringBuilder sb = new StringBuilder();
        // 遍历�?��更新的各个字段并写入where条件�?
        for (String updateField : upFields) {
            sb.append(updateField)
                    .append("=:")
                    .append(updateField)
                    .append(",");
        }
        sb.deleteCharAt(sb.length() - 1);
        return sb.toString();
    }

    /**
     * 设置更新列的�?
     *
     * @param map 从消息队列中读取的数�?
     * @param bs
     */
    private void setUpdateValue(Map map, BoundStatement bs, Set<String> feilds) {
        Set<String> f = new HashSet();
        f.addAll(feilds);
        f.remove("key");
        // 遍历�?��更新的各个列并分别设置更新�?
        for (String updateField : f) {
            Type type = columnMap.get(updateField);
            Object val = map.get(updateField);
            type.bind(updateField, val, bs);
        }
    }

    private String getUpdatesql(String querykey, Set<String> feilds) {
        Set<String> f = new HashSet();
        f.addAll(feilds);
        f.remove("key");
        String[] querykeys = querykey.split(":");
        int length = querykey.split(":").length;
        String sql = "update " + tableName + " set "
                + getUpdateStr(f) + " where ";
        StringBuilder sb = new StringBuilder();
        sb.append(sql);
        int i = 0;
        for (String s : querykeys) {
            sb.append(s);
            i++;
            sb.append("=:" + "key" + " and ");
        }
//        sb.append("seqn=:" + "seqn" + " and ");
        int j = sb.lastIndexOf(" and");
        String ss2 = sb.substring(0, j) + " if exists";
        return ss2;
    }

    private void setKeys(String querykey, Map m, BoundStatement bs) {
        int idx = 0;
        String[] querykeys = querykey.split(":");
        int i = 0;
        for (String s : querykeys) {
            i++;
            columnMap.get(s).bind("key" + i, m.get(s), bs);
        };
    }

}
